package flink_p2_sql

import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.table.api.Table
import org.apache.flink.table.api.scala.StreamTableEnvironment
import org.apache.flink.types.Row

object sql02_window {


  /**
   * 使用sqlapi + window
   * @param args
   */
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val tableEnv: StreamTableEnvironment = StreamTableEnvironment.create(env)

    import org.apache.flink.streaming.api.scala._
    import org.apache.flink.table.api.scala._

    val dataStream: DataStream[(Long, String, Int)] = env.socketTextStream("node1", 8889)
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[String](Time.seconds(2)) {
        override def extractTimestamp(element: String): Long = element.split(" ")(0).toLong
      })
      .map(data => {
        val arr: Array[String] = data.split(" ")
        (arr(0).toLong, arr(1), arr(2).toInt)
      })


//    tableEnv.registerDataStream("t_user", dataStream,'time.rowtime, 'name, 'age)
    val userTable: Table = tableEnv.fromDataStream(dataStream, 'event_time.rowtime, 'name, 'age)


    //test1: 每经过5秒 统计最近5秒内各年龄的人数（滚动窗口）
//    val resTable: Table = tableEnv.sqlQuery(s"select age,count(*) from ${userTable} group by tumble(event_time,interval '5' second),age")


    //test2:  每经过5秒统计最近10秒内各年龄人数(滑动窗口)
    //hop(三个参数)：第一个参数：时间字段,第二个：滑动步长，第三个:窗口长度
    val resTable: Table = tableEnv.sqlQuery(s"select age,count(*) from ${userTable} group by hop(event_time,interval '5' second, interval '10' second),age")



    resTable.toRetractStream[Row].print()




    tableEnv.execute("sql api.")
  }
}
